using System;
using System.Collections;
using System.Collections.Concurrent;
using System.Collections.Generic;
using Me.Prettyprint.Cassandra.Connection.Client;
using Me.Prettyprint.Cassandra.Connection.Factory;
using Me.Prettyprint.Cassandra.Service;
using Me.Prettyprint.Cassandra.Utils;
using Counter = Me.Prettyprint.Cassandra.Service.CassandraClientMonitor.Counter;
using Me.Prettyprint.Hector.Api;
using Me.Prettyprint.Hector.Api.Exceptions;
using Org.Apache.Cassandra.Thrift;

namespace Me.Prettyprint.Cassandra.Connection
{
	/// <summary>
	/// HConnectionManager is responsible for retrying queries in the event
    /// a host has issues. It waits until operateWithFailover is called to
    /// get a connection from the host connection pool. This is done to 
    /// make it more efficient and not tie up a connection longer than
    /// necessary. 
	/// </summary>
	public class HConnectionManager
	{
		private Logger logger = new Logger(typeof(HConnectionManager));
		private ConcurrentDictionary<CassandraHost, HClientPool> hostPools;
		private ConcurrentDictionary<CassandraHost, HClientPool> suspendedHostPools;
		private ICollection<HClientPool> hostPoolValues;
		private CassandraHostRetryService cassandraHostRetryService;
		private NodeAutoDiscoverService nodeAutoDiscoverService;
		private LoadBalancingPolicy loadBalancingPolicy;
		private CassandraHostConfigurator cassandraHostConfigurator;
		private HClientFactory clientFactory;
		private HostTimeoutTracker hostTimeoutTracker;
		private ClockResolution clock;
		
		internal ExceptionsTranslator exceptionsTranslator;
		private CassandraClientMonitor monitor;
		private ConnectionManagerListenersHandler listenerHandler = new ConnectionManagerListenersHandler();
		
		public HConnectionManager(string clusterName, CassandraHostConfigurator cassandraHostConfigurator)
		{
			clientFactory = HClientFactoryProvider.createFactory(cassandraHostConfigurator);
			
			loadBalancingPolicy = cassandraHostConfigurator.LoadBalancingPolicy;
			clock = Me.Prettyprint.Cassandra.Service.CassandraHostConfigurator.GetClockResolution();
			hostPools = new ConcurrentDictionary<CassandraHost,HClientPool>();
			suspendedHostPools = new ConcurrentDictionary<CassandraHost,HClientPool>();
			ClusterName = clusterName;
            ConfigureLogLevel(cassandraHostConfigurator);
			if (cassandraHostConfigurator.RetryDownedHosts)
			{
				cassandraHostRetryService = new CassandraHostRetryService(this, clientFactory, cassandraHostConfigurator, listenerHandler);
			}
			foreach(CassandraHost host in cassandraHostConfigurator.buildCassandraHosts())
			{
				try
				{
					HClientPool hcp = loadBalancingPolicy.CreateConnection(clientFactory, host, monitor);
					hostPools.AddOrUpdate(host,hcp, (key,oldValue) => hcp);
				}
				catch (HectorTransportException hte)
				{
					logger.Error(string.Format("Could not start connection pool for host {0} - {1}", host, hte.Message));
					listenerHandler.FireOnHostDown(host);
					if (cassandraHostRetryService != null)
					{
						cassandraHostRetryService.Add(host);
					}
				}
			}
			
			if (cassandraHostConfigurator.UseHostTimeoutTracker)
			{
				hostTimeoutTracker = new HostTimeoutTracker(this, cassandraHostConfigurator);
			}
			exceptionsTranslator = new ExceptionsTranslatorImpl();
			this.cassandraHostConfigurator = cassandraHostConfigurator;
			hostPoolValues = hostPools.Values;
			if (cassandraHostConfigurator.AutoDiscoverHosts)
			{
				nodeAutoDiscoverService = new NodeAutoDiscoverService(this, cassandraHostConfigurator);
			}
			Timer = cassandraHostConfigurator.OpTimer;
		}
        public void SetCassandraClientMonitor(CassandraClientMonitor cmon)
        {
            this.monitor = cmon;
        }
        protected void ConfigureLogLevel(CassandraHostConfigurator cassandraHostConfigurator)
        {
            string llv = cassandraHostConfigurator.LogLevel;
            switch (llv)
            {
                case "Info":
                    Diagnostics.CassandraTraceSwitch.Level = System.Diagnostics.TraceLevel.Info;
                    break;
                case "Error":
                    Diagnostics.CassandraTraceSwitch.Level = System.Diagnostics.TraceLevel.Error;
                    break;
                case "Warn":
                    Diagnostics.CassandraTraceSwitch.Level = System.Diagnostics.TraceLevel.Warning;
                    break;
                case "Verbose":
                    Diagnostics.CassandraTraceSwitch.Level = System.Diagnostics.TraceLevel.Verbose;
                    break;
                default:
                    Diagnostics.CassandraTraceSwitch.Level = System.Diagnostics.TraceLevel.Info;
                    break;
            }
        }
		public HOpTimer Timer
		{
			get;
			set;
		}
		public string ClusterName
		{
			get;
            protected set;
		}
		public int CassandraHostRetryDelay
		{
			set
			{
				cassandraHostRetryService.RetryDelayInSeconds = value;
				cassandraHostRetryService.ApplyRetryDelay();
			}
		}
        public HClientPool ClientFromLBPolicy(List<CassandraHost> excludedHosts)
        {
            if (hostPools.IsEmpty)
            {
                throw new HectorException("All host pools marked down. Retry burden pushed out to client.");
            }
            hostPoolValues = hostPools.Values;
            return loadBalancingPolicy.GetPool(hostPoolValues, excludedHosts); ;
        }
		public void DoAddNodes()
		{
			new NodeDiscovery(cassandraHostConfigurator, this).AddNodes();
		}
        private void DoTimeoutCheck(CassandraHost cassandraHost)
        {
            if (hostTimeoutTracker != null && hostPools.Count > 1)
            {
                if (hostTimeoutTracker.CheckTimeout(cassandraHost))
                {
                    SuspendCassandraHost(cassandraHost);
                }
            }
        }
        public List<CassandraHost> GetHosts()
        {
            return new List<CassandraHost>(hostPools.Keys);
        }
        public List<HClientPool> ActivePools()
        {
            return new List<HClientPool>(hostPools.Values);
        }
        public List<string> StatusPerPool()
        {
            List<string> stats = new List<string>();
            foreach (HClientPool clientPool in hostPools.Values)
            {
                stats.Add(clientPool.StatusAsString);
            }
            return stats;
        }
        public List<CassandraHost> SuspendedCassandraHosts()
        {
            return new List<CassandraHost>(this.suspendedHostPools.Keys);
        }
		public bool AddCassandraHost(CassandraHost cassandraHost)
		{
			if (!hostPools.ContainsKey(cassandraHost))
			{
				HClientPool pool = null;
				try
				{
					cassandraHostConfigurator.ApplyConfig(cassandraHost);
					pool = cassandraHostConfigurator.LoadBalancingPolicy.CreateConnection(clientFactory, cassandraHost, monitor);
					hostPools.AddOrUpdate(cassandraHost, pool, (key, oldValue) => pool);
					logger.Info(string.Format("Added host {0} to pool", cassandraHost.Name));
					listenerHandler.FireOnAddHost(cassandraHost, true, null, null);
                    if (cassandraHostRetryService != null)
                    {
                        cassandraHostRetryService.Remove(cassandraHost);
                    }
					return true;
				}
				catch (HectorTransportException hte)
				{
					System.String errorMessage = "Transport exception host to HConnectionManager: " + cassandraHost;
					logger.Error(errorMessage, hte);
					listenerHandler.FireOnAddHost(cassandraHost, false, errorMessage, hte);
				}
				catch (System.Exception ex)
				{
					System.String errorMessage = "General exception host to HConnectionManager: " + cassandraHost;
					logger.Error(errorMessage, ex);
					listenerHandler.FireOnAddHost(cassandraHost, false, errorMessage, ex);
				}
			}
			else
			{
				System.String message = "Host already existed for pool " + cassandraHost.Name;
				logger.Info(message);
				listenerHandler.FireOnAddHost(cassandraHost, false, message, null);
			}
			return false;
		}
		public bool RemoveCassandraHost(CassandraHost cassandraHost)
		{
			bool removed = hostPools.ContainsKey(cassandraHost);
			string message;
			if (removed)
			{
				HClientPool pool = null;
                hostPools.TryRemove(cassandraHost, out pool);
				message = "Removed from hostPools";
				if (pool == null)
				{
					logger.Info(string.Format("removeCassandraHost looking for host {0} in suspendedHostPools", cassandraHost));
					suspendedHostPools.TryRemove(cassandraHost, out pool);
					message = "Removed from suspendedHostPools";
				}
				if (pool != null)
				{
					pool.Shutdown();
				}
				else
				{
					removed = false;
					message = "Removed by another thread";
					logger.Info(string.Format("removeCassandraHost attempt miss for CassandraHost {0} May have been beaten by another thread?", cassandraHost));
				}
			}
			else if (cassandraHostRetryService != null && cassandraHostRetryService.Contains(cassandraHost))
			{
				logger.Info(string.Format("Host {} not in active pools, but found in retry service.", cassandraHost));
				removed = cassandraHostRetryService.Remove(cassandraHost);
				message = "Removed from Downed hosts";
			}
			else
			{
				message = "Host not found";
				logger.Info(string.Format("Remove requested on a host that was not found in active or disabled pools: {0}", cassandraHost));
			}
			logger.Info(string.Format("Remove status for CassandraHost pool {0} was {1}", cassandraHost, removed));
			listenerHandler.FireOnRemoveHost(cassandraHost, removed, message);
			return removed;
		}
		public bool SuspendCassandraHost(CassandraHost cassandraHost)
		{
            try
            {
                HClientPool pool = null;
                hostPools.TryRemove(cassandraHost, out pool);
                bool removed = pool != null;
                if (removed)
                {
                    suspendedHostPools.AddOrUpdate(cassandraHost, pool, (key, oldvalue) => pool);
                }
                listenerHandler.FireOnSuspendHost(cassandraHost, removed);
                logger.Info(string.Format("Suspend operation status was {0} for CassandraHost {1}", removed, cassandraHost));
                return removed;
            }
            catch (Exception e)
            {
                logger.Warning(e.Message);
                return false;
            }
		}
		public bool UnsuspendCassandraHost(CassandraHost cassandraHost)
		{
            try
            {
                HClientPool pool = null;
                suspendedHostPools.TryRemove(cassandraHost, out pool);
                bool readded = pool != null;
                if (readded)
                {
                    bool alreadyThere = hostPools.ContainsKey(cassandraHost);
                    if (alreadyThere)
                    {
                        logger.Error(string.Format("Unsuspend called on a pool that was already active for CassandraHost {}", cassandraHost));
                        pool.Shutdown();
                    }
                    else
                    {
                        hostPools.TryAdd(cassandraHost, pool);
                    }
                    return alreadyThere;
                }
                listenerHandler.FireOnUnSuspendHost(cassandraHost, readded);
                logger.Info(string.Format("UN-Suspend operation status was {} for CassandraHost {}", readded, cassandraHost));
                return readded;
            }
            catch (Exception e)
            {
                logger.Warning(e.Message);
                return false;
            }
		}
        public List<CassandraHost> GetDownedHost()
        {
            return this.cassandraHostRetryService.GetDownedHosts();
        }
        public List<HClientPool> GetActivePools()
        {
            return new List<HClientPool>(hostPools.Values);
        }
		private void CloseClient(HClient client)
		{
			if (client != null)
			{
				client.Close();
				client.ClearAuthentication();
			}
		}
		public void AddListener(string listenerName, ConnectionManagerListener listener)
		{
			listenerHandler.Add(listenerName, listener);
		}
		public void RemoveListener(string listenerName)
		{
			listenerHandler.Remove(listenerName);
		}
		public void RemoveAllListeners()
		{
			listenerHandler.Clear();
		}
		private void TimeoutCheck(CassandraHost cassandraHost)
		{
			if (hostTimeoutTracker != null && hostPools.Count > 1)
			{
				if (hostTimeoutTracker.CheckTimeout(cassandraHost))
				{
					SuspendCassandraHost(cassandraHost);
				}
			}
		}
		private void SleepBetweenHostSkips(FailoverPolicy failoverPolicy)
		{
            if (failoverPolicy.GetSleepBetweenHostsMS() > 0)
			{
				try
				{
                    System.Threading.Thread.Sleep(new System.TimeSpan((System.Int64)10000 * failoverPolicy.GetSleepBetweenHostsMS()));
				}
				catch (System.Threading.ThreadInterruptedException e)
				{
					logger.Warning(string.Format("Sleep between hosts interrupted {0}", e));
				}
			}
		}
		internal void ReleaseClient(HClient client)
		{
			if (client == null)
            {
				return ;
            }
            HClientPool pool = null;
            hostPools.TryGetValue(client.CassandraHost, out pool);
			if (pool == null)
			{
				suspendedHostPools.TryGetValue(client.CassandraHost, out pool);
			}
			if (pool != null)
			{
				pool.ReleaseClient(client);
			}
			else
			{
				logger.Info(string.Format("Client {0} released to inactive or dead pool. Closing.", client));
				client.Close();
			}
		}
		internal HClient BorrowClient()
		{
			HClientPool pool = ClientFromLBPolicy(null);
			if (pool != null)
			{
				return pool.BorrowClient();
			}
			return null;
		}
		internal void MarkHostAsDown(CassandraHost cassandraHost)
		{
			logger.Error(string.Format("MARK HOST AS DOWN TRIGGERED for host {0}", cassandraHost.Name));
			listenerHandler.FireOnHostDown(cassandraHost);
			HClientPool pool = null;
            hostPools.TryGetValue(cassandraHost, out pool);
			if (pool != null)
			{
                if (this.cassandraHostRetryService != null)
                {
                    this.cassandraHostRetryService.Add(cassandraHost);
                }
                logger.Error(string.Format("Pool state on shutdown: {0}", pool.StatusAsString));
                pool.Shutdown();
            }
            hostPools.TryRemove(cassandraHost, out pool);
        }
		public long CreateClock()
		{
			return this.clock.CreateClock();
		}
		public void Shutdown()
		{
			logger.Info("Shutdown called on HConnectionManager");
			if (cassandraHostRetryService != null)
            {
				cassandraHostRetryService.Shutdown();
            }
			if (nodeAutoDiscoverService != null)
            {
				nodeAutoDiscoverService.Shutdown();
            }
			if (hostTimeoutTracker != null)
            {
				hostTimeoutTracker.Shutdown();
            }
			foreach (HClientPool pool in hostPools.Values)
			{
				try
				{
					pool.Shutdown();
				}
				catch (System.ArgumentException e)
				{
					logger.Error(string.Format("Out of order in HConnectionManager shutdown()?: {0}", e.Message));
				}
			}
		}
        public void operateWithFailover<T>(Me.Prettyprint.Cassandra.Service.Operation<T> op)
        {
            object timerToken = Timer.Start(op.StopWatchTagName);
            int retries = 0;
            retries = Math.Min(op.FailoverPolicy.NumRetries, hostPools.Count);
            HClient client = null;
            HClientPool pool = null;
            bool success = false;
            bool retryable = false;
            List<CassandraHost> excludeHosts = new List<CassandraHost>();
            while (!success)
            {
                try
                {
                    pool = ClientFromLBPolicy(excludeHosts);
                    client = pool.BorrowClient();
                    if (op.Credentials != null && op.Credentials.Count > 0 && !client.AlreadyAuthenticated(op.Credentials))
                    {
                        AuthenticationRequest request = new AuthenticationRequest();
                        request.Credentials = op.Credentials;
                        client.GetCassandra().login(request);
                        client.SetAuthentication(op.Credentials);
                    }
                    Org.Apache.Cassandra.Thrift.Cassandra.Client c = client.GetCassandra(op.KeyspaceName);
                    op.ExecuteAndSetResult(c, pool.CassandraHost);
                    success = true;
                    client.UpdateLastSuccessTime();
                    Timer.Stop(timerToken, op.StopWatchTagName, true);
                    break;
                }
                catch (Exception e)
                {
                    // in case exception occurs, we retry with different node
                    HectorException he = exceptionsTranslator.translate(e);
                    if (he is HUnavailableException)
                    {
                        throw he;
                    }
                    else if (he is HInvalidRequestException || he is HCassandraInternalException)
                    {
                        CloseClient(client);
                        throw he;
                    }
                    else if (he is HectorTransportException)
                    {
                        CloseClient(client);
                        MarkHostAsDown(pool.CassandraHost);
                        excludeHosts.Add(pool.CassandraHost);
                        retryable = op.FailoverPolicy.ShouldRetryFor(typeof(HectorTransportException));
                        if (monitor != null)
                        {
                            monitor.IncrementCounter(Counter.RECOVERABLE_TRANSPORT_EXCEPTIONS);
                        }
                    }
                    else if (he is HTimedOutException)
                    {
                        DoTimeoutCheck(pool.CassandraHost);
                        retryable = op.FailoverPolicy.ShouldRetryFor(typeof(HTimedOutException));
                        if (monitor != null)
                        {
                            monitor.IncrementCounter(Counter.RECOVERABLE_TIMED_OUT_EXCEPTIONS);
                        }
                        client.Close();
                    }
                    else if (he is HPoolRecoverableException)
                    {
                        retryable = op.FailoverPolicy.ShouldRetryFor(typeof(HPoolRecoverableException));
                        if (hostPools.Count == 1)
                        {
                            throw he;
                        }
                        if (monitor != null)
                        {
                            monitor.IncrementCounter(Counter.POOL_EXHAUSTED);
                        }
                        excludeHosts.Add(pool.CassandraHost);
                    }
                    else
                    {
                        retryable = false;
                    }
                    if (retries <= 0 || retryable == false)
                    {
                        throw he;
                    }
                    logger.Warning(string.Format("Could not fullfill request on this host {0}", client));
                    logger.Warning(string.Format("Exception: {0}", he));
                    if (monitor != null)
                    {
                        monitor.IncrementCounter(Counter.SKIP_HOST_SUCCESS);
                    }
                    SleepBetweenHostSkips(op.FailoverPolicy);
                }
                finally
                {
                    if (!success)
                    {
                        if (monitor != null)
                        {
                            monitor.IncrementCounter(op.FailCounter);
                        }
                        Timer.Stop(timerToken, op.StopWatchTagName, false);
                    }
                    ReleaseClient(client);
                    client = null;
                }
            }
        }
	}
}